package com.navinfo.tripanalysis.service.impl;

import com.navinfo.tripanalysis.pojo.*;
import com.navinfo.tripanalysis.service.AbstractLoad0200DataServiceBigData;
import com.navinfo.tripanalysis.util.BigDataUtils;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;

import java.util.List;

/**
 * 从大数据平台加载0200服务
 *
 * @author 沈东生
 */
public class Load0200DataServiceHdfsImpl extends AbstractLoad0200DataServiceBigData {
    private String basePath;
    public String getBasePath() {
        return basePath;
    }
    public void setBasePath(String basePath) {
        this.basePath = basePath;
    }

    @Override
    public JavaRDD<Point> load(SparkSession spark, JavaSparkContext jsc, LoadDataParam param) {
        List<String> realPaths = BigDataUtils.getHDFSPath(param.getDay(), (List<Integer>)param.getExtendParam("batchList"), getBasePath());
        if (realPaths.size() > 0) {
            return toJavaRdd(spark.read().parquet(realPaths.toArray(new String[realPaths.size()])).select("terminalid", COL_NAMES.toArray(new String[COL_NAMES.size()])).where(getWhereSql0200(param)));
        }
        return null;
    }


}
